package com.grain.mall.goods.mq;

import base.MqBaseModel;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import myConst.MQConst;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.io.IOException;
import java.util.List;

public class Consumer {

    public static void main(String[] args) throws Exception {
        MQConst.MQEnum mqEnum = MQConst.MQEnum.GOODS_ORDER_CREATE_LIST;
        // 初始化consumer，并设置consumer group name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqEnum.getConsumerGroup());
        // 设置NameServer地址
        consumer.setNamesrvAddr("192.168.171.129:9876");
        //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
        consumer.subscribe(mqEnum.getTopic(), "*");
//        consumer.subscribe(mqEnum.getTopic(), "TagB");
        //注册回调接口来处理从Broker中收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgs.stream().forEach(item -> {
                    String tags = item.getTags();
                    System.out.println(tags);
                    String bodyStr = new String(item.getBody());
                    MqBaseModel<String> mqBaseModel = JSONObject.parseObject(bodyStr, MqBaseModel.class);
                    System.out.println(mqBaseModel.getSendTime() + "==========" + DateUtil.now());
                    System.out.println(mqBaseModel);
                });
                // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动Consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
        System.in.read();
    }
}
